package org.shuzhou.d_stateful;

import org.apache.flink.api.common.accumulators.LongCounter;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


/**
 * 数据集：data/people.txt
 * 使用内置计数器LongCounter统计总人数，而非使用keyed state，后者要求必须分区
 */
public class PersonInnerCounter {
    public static void main(String[] args) throws Exception {

        // set up the streaming execution environment
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        /**
         * 获取数据，数据源为Kafka，避免从端口获取中文数据乱码
         * 数据格式
         * 王矗馨,女,汉族,411424195001010028,19500101,150977754190,kk4jlw850x@sina.com,甘肃省临夏回族自治州和政县CNR路884号
         */
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("192.168.56.151:9092,192.168.56.152:9092,192.168.56.153:9092")
                .setTopics("persion")
                .setGroupId("root")
                .setStartingOffsets(OffsetsInitializer.earliest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        DataStream<String> people = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");

        /**
         * 获取数据，数据源为kafka
         * 数据格式
         * 王矗馨,女,汉族,411424195001010028,19500101,150977754190,kk4jlw850x@sina.com,甘肃省临夏回族自治州和政县CNR路884号
         * 直接统计目前已经累加的条数
         */
        SingleOutputStreamOperator<Long> ds = people.map(new RichMapFunction<String, Long>() {
            
            private transient LongCounter counter;

            // 初始化内置计数器
            @Override
            public void open(Configuration parameters) throws Exception {
                // 获取计数器
                counter = getRuntimeContext().getLongCounter("myCounter");
            }

            @Override
            public Long map(String value) throws Exception {
                // 计数器自增
                counter.add(1);
                return counter.getLocalValue();
            }
        });

        ds.print();

        env.execute("Flink Operator Example");
    }
}
